package cn.gwm.flink.streaming.ods;

import cn.gwm.flink.streaming.constant.VehicleType;
import cn.gwm.flink.streaming.function.map.GpsToRedisFunction;
import cn.gwm.flink.streaming.ods.model.ModelCanEnum;
import cn.gwm.flink.streaming.ods.model.ModelGpsEnum;
import cn.gwm.flink.streaming.ods.model.StandardModelConf;
import cn.gwm.utils.ConfigLoader;
import cn.hutool.json.JSONObject;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

/**
 * @ClassName OdsHandle
 * @Author xzh
 * @Date 2023/3/3 13:34
 **/
public class OdsHandle {

    private static final int minLenEs13 = 800;
    private static final String CAN_PREFIX_ES13 = "ods_can_es13";
    private static final String GPS_PREFIX_ES13 = "ods_gps_es13";
    /**
     *@Author xzh
     *@Description es13车型数据处理
     *@Date 2023/3/3 13:56
     *@Param [canProcess, gpsProcess, canTagMap, gpsTagMap, bootActive]
     *@return void
     **/
    public static void handleEs13(SingleOutputStreamOperator<String> canProcess, SingleOutputStreamOperator<String> gpsProcess,
                              Map<String, OutputTag<String>> canTagMap, Map<String, OutputTag<String>> gpsTagMap,String bootActive){
        ModelHandle handle = new ModelHandle(ModelCanEnum.ES13, ModelGpsEnum.ES13, minLenEs13, VehicleType.ES13,
                CAN_PREFIX_ES13, GPS_PREFIX_ES13, VehicleType.POWER_TYPE_EV, ConfigLoader.get("kafka.clean.topic.ods.es13"),
                ConfigLoader.get("hbase.ods.table.name.gps.es13"));
        //整车分析车型，用过滤字段的
        SingleOutputStreamOperator<JSONObject> stream = handle.connectStream(canProcess, gpsProcess, canTagMap, gpsTagMap);
        handle.addCanToHive(stream);
//        //gps 数据处理
        SingleOutputStreamOperator<JSONObject> gpsStream = handle.getGpsStream(gpsProcess, gpsTagMap);
        handle.addGpsToHive(gpsStream);
        SingleOutputStreamOperator<JSONObject> hbaseStream
                = gpsStream.flatMap(new GpsToRedisFunction(VehicleType.ES13)).name("flatMap" + VehicleType.ES13);
        handle.addGpsToHbase(hbaseStream);
        if("dev".equals(bootActive)){
            SingleOutputStreamOperator<JSONObject> canStream = handle.getCanStream(canProcess, canTagMap,minLenEs13);
            handle.addCanToDefaultKafka(canStream);
        }
    }

    private static final int minLenV71 = 800;
    private static final String CAN_PREFIX_V71 = "ods_can_v71";
    private static final String GPS_PREFIX_V71 = "ods_gps_v71";
    /**
     *@Author xzh
     *@Description v71phev车型数据处理
     *@Date 2023/3/3 13:56
     *@Param [canProcess, gpsProcess, canTagMap, gpsTagMap, bootActive]
     *@return void
     **/
    public static void handleV71(SingleOutputStreamOperator<String> canProcess, SingleOutputStreamOperator<String> gpsProcess,
                                 Map<String, OutputTag<String>> canTagMap, Map<String, OutputTag<String>> gpsTagMap,String bootActive){
        ModelHandle handle = new ModelHandle(ModelCanEnum.V71_PHEV, ModelGpsEnum.V71_PHEV, minLenV71, VehicleType.V71_PHEV,
                CAN_PREFIX_V71, GPS_PREFIX_V71, VehicleType.POWER_TYPE_PHEV, ConfigLoader.get("kafka.clean.topic.ods.v71"),
                ConfigLoader.get("hbase.ods.table.name.gps.v71"));
        //整车分析车型，用过滤字段的
        SingleOutputStreamOperator<JSONObject> stream = handle.connectStream(canProcess, gpsProcess, canTagMap, gpsTagMap);
        handle.addCanToHive(stream);
        //gps 数据处理
        SingleOutputStreamOperator<JSONObject> gpsStream = handle.getGpsStream(gpsProcess, gpsTagMap);
        handle.addGpsToHive(gpsStream);
        SingleOutputStreamOperator<JSONObject> hbaseStream
                = gpsStream.flatMap(new GpsToRedisFunction(VehicleType.V71_PHEV)).name("flatMap"+VehicleType.V71_PHEV);
        handle.addGpsToHbase(hbaseStream);
        if("dev".equals(bootActive)){
            SingleOutputStreamOperator<JSONObject> canStream = handle.getCanStream(canProcess, canTagMap);
            handle.addCanToDefaultKafka(canStream);
        }
    }
    private static final int minLenB01 = 500;
    private static final String CAN_PREFIX_B01 = "ods_can_b01";
    private static final String GPS_PREFIX_B01 = "ods_gps_b01";
    /**
     *@Author xzh
     *@Description b01phev车型数据处理
     *@Date 2023/3/3 13:56
     *@Param [canProcess, gpsProcess, canTagMap, gpsTagMap, bootActive]
     *@return void
     **/
    public static void handleB01(SingleOutputStreamOperator<String> canProcess, SingleOutputStreamOperator<String> gpsProcess,
                                 Map<String, OutputTag<String>> canTagMap, Map<String, OutputTag<String>> gpsTagMap){
        ModelHandle handle = new ModelHandle(ModelCanEnum.B01_PHEV, ModelGpsEnum.B01_PHEV, minLenB01, VehicleType.B01_PHEV,
                CAN_PREFIX_B01, GPS_PREFIX_B01, VehicleType.POWER_TYPE_PHEV);
        //gps 数据处理
        SingleOutputStreamOperator<JSONObject> gpsStream = handle.getGpsStream(gpsProcess, gpsTagMap);
        handle.addGpsToHive(gpsStream);
        SingleOutputStreamOperator<JSONObject> canStream = handle.getCanStream(canProcess, canTagMap,minLenB01);
        handle.addCanToDefaultKafka(canStream);
        handle.addCanToHive(canStream);

    }

    private static final int minLenB07 = 600;
    private static final String CAN_PREFIX_B07 = "ods_can_b07";
    private static final String GPS_PREFIX_B07 = "ods_gps_b07";
    /**
     *@Author xzh
     *@Description b01phev车型数据处理
     *@Date 2023/3/3 13:56
     *@Param [canProcess, gpsProcess, canTagMap, gpsTagMap, bootActive]
     *@return void
     **/
    public static void handleB07(SingleOutputStreamOperator<String> canProcess, SingleOutputStreamOperator<String> gpsProcess,
                                 Map<String, OutputTag<String>> canTagMap, Map<String, OutputTag<String>> gpsTagMap){
        ModelHandle handle = new ModelHandle(ModelCanEnum.B07_PHEV, ModelGpsEnum.B07_PHEV, minLenB07, VehicleType.B07_PHEV,
                CAN_PREFIX_B07, GPS_PREFIX_B07, VehicleType.POWER_TYPE_PHEV);
        //gps 数据处理
        SingleOutputStreamOperator<JSONObject> gpsStream = handle.getGpsStream(gpsProcess, gpsTagMap);
        handle.addGpsToHive(gpsStream);
        SingleOutputStreamOperator<JSONObject> canStream = handle.getCanStream(canProcess, canTagMap,minLenB07);
        handle.addCanToDefaultKafka(canStream);
        handle.addCanToHive(canStream);

    }

    private static final int minLenC01 = 850;
    private static final String CAN_PREFIX_C01 = "ods_can_c01";
    private static final String GPS_PREFIX_C01 = "ods_gps_c01";
    /**
     *@Author xzh
     *@Description b01phev车型数据处理
     *@Date 2023/3/3 13:56
     *@Param [canProcess, gpsProcess, canTagMap, gpsTagMap, bootActive]
     *@return void
     **/
    public static void handleC01(SingleOutputStreamOperator<String> canProcess, SingleOutputStreamOperator<String> gpsProcess,
                                 Map<String, OutputTag<String>> canTagMap, Map<String, OutputTag<String>> gpsTagMap){
        ModelHandle handle = new ModelHandle(ModelCanEnum.C01_PHEV, ModelGpsEnum.C01_PHEV, minLenC01, VehicleType.C01_PHEV,
                CAN_PREFIX_C01, GPS_PREFIX_C01, VehicleType.POWER_TYPE_PHEV);
        //gps 数据处理
        SingleOutputStreamOperator<JSONObject> gpsStream = handle.getGpsStream(gpsProcess, gpsTagMap);
        handle.addGpsToHive(gpsStream);
        SingleOutputStreamOperator<JSONObject> canStream = handle.getCanStream(canProcess, canTagMap,minLenC01);
        handle.addCanToDefaultKafka(canStream);
        handle.addCanToHive(canStream);

    }

    private static final int minLenEc24 = 800;
    private static final String CAN_PREFIX_EC24 = "ods_can_ec24";
    private static final String GPS_PREFIX_EC24 = "ods_gps_ec24";
    /**
     *@Author xzh
     *@Description ec24车型数据处理
     *@Date 2023/3/3 13:56
     *@Param [canProcess, gpsProcess, canTagMap, gpsTagMap, bootActive]
     *@return void
     **/
    public static void handleEc24(SingleOutputStreamOperator<String> canProcess, SingleOutputStreamOperator<String> gpsProcess,
                                  Map<String, OutputTag<String>> canTagMap, Map<String, OutputTag<String>> gpsTagMap){
        ModelHandle handle = new ModelHandle(ModelCanEnum.EC24, ModelGpsEnum.EC24, minLenEc24, VehicleType.EC24,
                CAN_PREFIX_EC24, GPS_PREFIX_EC24, VehicleType.POWER_TYPE_EV);
        //gps 数据处理
        SingleOutputStreamOperator<JSONObject> gpsStream = handle.getGpsStream(gpsProcess, gpsTagMap);
        handle.addGpsToHive(gpsStream);
        SingleOutputStreamOperator<JSONObject> canStream = handle.getCanStream(canProcess, canTagMap,minLenEc24);
        handle.addCanToHive(canStream);
        //检测车型，全字段数据获取
        handle.addCanToDefaultKafka(canStream);
    }

    private static final int minLenEs24 = 400;
    private static final String CAN_PREFIX_ES24 = "ods_can_es24";
    private static final String GPS_PREFIX_ES24 = "ods_gps_es24";
    /**
     *@Author xzh
     *@Description es24车型数据处理
     *@Date 2023/3/3 13:56
     *@Param [canProcess, gpsProcess, canTagMap, gpsTagMap, bootActive]
     *@return void
     **/
    public static void handleEs24(SingleOutputStreamOperator<String> canProcess, SingleOutputStreamOperator<String> gpsProcess,
                                  Map<String, OutputTag<String>> canTagMap, Map<String, OutputTag<String>> gpsTagMap){
        ModelHandle handle = new ModelHandle(ModelCanEnum.ES24, ModelGpsEnum.ES24, minLenEs24, VehicleType.ES24,
                CAN_PREFIX_ES24, GPS_PREFIX_ES24, VehicleType.POWER_TYPE_EV);
        //gps 数据处理
        SingleOutputStreamOperator<JSONObject> gpsStream = handle.getGpsStream(gpsProcess, gpsTagMap);
        handle.addGpsToHive(gpsStream);
        SingleOutputStreamOperator<JSONObject> canStream = handle.getCanStream(canProcess, canTagMap,minLenEs24);
        handle.addCanToDefaultKafka(canStream);
        handle.addCanToHive(canStream);
    }


    private static final int minLenP03 = 500;
    private static final String CAN_PREFIX_P03 = "ods_can_p03";
    private static final String GPS_PREFIX_P03 = "ods_gps_p03";
    /**
     *@Author xzh
     *@Description P03_PHEV车型数据处理
     *@Date 2023/3/3 13:56
     *@Param [canProcess, gpsProcess, canTagMap, gpsTagMap, bootActive]
     *@return void
     **/
    public static void handleP03(SingleOutputStreamOperator<String> canProcess, SingleOutputStreamOperator<String> gpsProcess,
                                 Map<String, OutputTag<String>> canTagMap, Map<String, OutputTag<String>> gpsTagMap,String bootActive){
        ModelHandle handle = new ModelHandle(ModelCanEnum.P03_PHEV, ModelGpsEnum.P03_PHEV, minLenP03, VehicleType.P03_PHEV,
                CAN_PREFIX_P03, GPS_PREFIX_P03, VehicleType.POWER_TYPE_PHEV, ConfigLoader.get("kafka.clean.topic.ods.p03"),
                ConfigLoader.get("hbase.ods.table.name.gps.p03"));
        //整车分析车型，用过滤字段的
        SingleOutputStreamOperator<JSONObject> stream = handle.connectStream(canProcess, gpsProcess, canTagMap, gpsTagMap);
        handle.addCanToHive(stream);
        //gps 数据处理
        SingleOutputStreamOperator<JSONObject> gpsStream = handle.getGpsStream(gpsProcess, gpsTagMap);
        handle.addGpsToHive(gpsStream);
        SingleOutputStreamOperator<JSONObject> hbaseStream
                = gpsStream.flatMap(new GpsToRedisFunction(VehicleType.P03_PHEV)).name("flatMap"+VehicleType.P03_PHEV);
        handle.addGpsToHbase(hbaseStream);
    }


    public static void handle(SingleOutputStreamOperator<String> process, Map<String, OutputTag<String>> canTagMap, List<StandardModelConf> confList) {
        for(StandardModelConf conf:confList){
            OdsModelHandle handle = new OdsModelHandle(conf);
            handle.connectStream(process, canTagMap);
        }
    }

    public static void handle2(SingleOutputStreamOperator<String> process, Map<String, OutputTag<String>> canTagMap, List<StandardModelConf> confList) {
        for(StandardModelConf conf:confList){
            OdsModelHandle handle = new OdsModelHandle(conf);
            handle.connectStream2(process, canTagMap);
        }
    }
}
